/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.gwxs;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.REAbstractSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.global.GlobalParas;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author lenovo-pc
 */
public class JsonGwxSender implements Runnable {

    private int addCount = 0;
    private long startTime = System.currentTimeMillis();
    static org.apache.log4j.Logger logger = null;

    static {
        logger = org.apache.log4j.Logger.getLogger(JsonGwxSender.class.getName());
    }

    ThreadPoolExecutor threadpool;

    public JsonGwxSender() {
        //创建一个线程池
        threadpool = new ThreadPoolExecutor(12, 12, 3, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(500),
                new ThreadFactory() {
            AtomicInteger idx = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "JsonGwxTask_" + idx.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("", ex);
                    }
                }
            }
        });
    }

    @Override
    public void run() {

        logger.info("It is running!");
        RESendSession session;
        ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        ReentrantReadWriteLock.ReadLock readlock = readWriteLock.readLock();
        ReentrantReadWriteLock.WriteLock writelock = readWriteLock.writeLock();
        SessionPool pool;
        try {
            REConnection conn = new REConnection(GlobalParas.IP_PORT);
            //logger.info("the rocketmq port is" +GlobalParas.LoadProperties("IP_PORT"));
            //builder
            RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(GlobalParas.MQ_GWX);
            builder.setTimeoutMillisec(10000);

            //交互的类型初始化
            for (String columntype : GlobalParas.TYPE_NEWS.split("\\|")) {
                switch (columntype.split("[:]")[1]) {
                    case "int":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Int, true);
                        break;
                    case "ints":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Ints, true);
                        break;
                    case "long":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Long, true);
                        break;
                    case "longs":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Longs, true);
                        break;
                    case "string":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.String, true);
                        break;
                    case "strings":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Strings, true);
                        break;
                    case "boolean":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Boolean, true);
                        break;
                    case "struct":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Struct, true);
                        REFieldType.Struct.setDetail(columntype.split("[:]")[0].toLowerCase(), new HashMap<String, REFieldType>());
                        break;
                    case "structs":
                        builder.addColumn(columntype.split("[:]")[0].toLowerCase(), REFieldType.Structs, true);
                        REFieldType.Structs.setDetail(columntype.split("[:]")[0].toLowerCase(), new HashMap<String, REFieldType>());
                        break;
                    default:
                        logger.error("The type is not found");
                }
            }
            pool = new SessionPool(builder);
            //session
//            session = (RESendSession) builder.build();
//
//            session.start();

        } catch (Exception ex) {
            logger.error("init session error :" + ex, ex);
            return;
        }

        String json = null;
        ArrayList<String> jsonList;
        while (true) {
            jsonList = new ArrayList<>(GlobalParas.SendNum);
            try {
                Long strSize = 0L;
                for (int i = 0; i < GlobalParas.SendNum; ++i) {
                    //logger.info("try "+i);
                    String str = GlobalParas.gwxQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (str == null) {
                        break;
                    }
                    strSize += str.length();
                    jsonList.add(str);
                }
                if (jsonList.isEmpty()) {
                    Thread.sleep(3000);
                    continue;
                }
                logger.debug("Poll Size is: " + jsonList.size() + " content size: " + strSize);
                strSize = 0L;
                threadpool.execute(new JsonGwxTask4(jsonList, pool));
                ++addCount;
            } catch (Exception ex) {
                logger.error("this user thread error :" + ex, ex);
            }
        }

    }

    static public class SessionPool_old {

        RESendSessionBuilder builder;
        LinkedList<RESendSession> pool;

        public SessionPool_old(RESendSessionBuilder builder) {
            this.builder = builder;
            pool = new LinkedList<>();
        }

        public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.removeFirst();
            } catch (NoSuchElementException ex) {
                RESendSession rtn = (RESendSession) builder.build();
                rtn.start();
                return rtn;
            }
        }

        public synchronized void returnSession(RESendSession session) {
            pool.add(session);
        }
    }

    static public class SessionPool {

        static private final int poolSize = 32;

        RESendSessionBuilder builder;
        LinkedBlockingQueue<RESendSession> pool;
        int currentMax;
        Lock maxLock;

        public SessionPool(RESendSessionBuilder builder) throws REConnectionException {
            this.builder = builder;
            pool = new LinkedBlockingQueue<>();
            currentMax = 0;
            maxLock = new ReentrantLock();
        }

        public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.remove();
            } catch (NoSuchElementException ex) {
                if (currentMax < poolSize) {
                    maxLock.lock();
                    try {
                        if (currentMax < poolSize) {
                            RESendSession rtn = (RESendSession) builder.build();
                            rtn.start();
                            ++currentMax;
                            return rtn;
                        }
                    } finally {
                        maxLock.unlock();
                    }
                }
                while (true) {
                    try {
                        return pool.take();
                    } catch (InterruptedException ex1) {
                    }
                }
            }
        }

        public synchronized void returnSession(RESendSession session) {
            pool.add(session);
        }
    }

}
